package myEddy.core;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.LinkedList;
import java.util.concurrent.PriorityBlockingQueue;

import myEddy.operator.Operator;
import myEddy.tuple.Tuple;


public class Eddy implements Runnable
{
	
        /** Attributes **/
        
        /**
         * Input queue. IOW, FIFO of tuples to deal with
         * Priority queue ?
         */
        private PriorityBlockingQueue<Tuple> input;
        
        /**
         * Result of eddy.
         */
        private LinkedList<Tuple> output;
        
        /**
         *  List of operators needed
         */
        private ArrayList<Operator> operators;
        
        /**
         * Corresponding selectivity based on tokens
         */
        private ArrayList<Integer> operators_selectivity;
        
        /**
         * Number of operator called
         */
        private int nbr_operator_call;
        
        /**
         * Tell if the eddy should be running or not (loop)
         */
        private boolean process;
        
        /**
         * Minimum number of operation before doing analytic scheduling
         */
        private final int min_nbr_op = 20;
        
        /**
         * Selected scheduling method
         */
        int sel_sche_met;
        
        /** Methods **/
        
        /**
         * Constructors
         */
        public Eddy()
        {
                input = new PriorityBlockingQueue<Tuple>(1, new TupleComparator());
                output = new LinkedList<Tuple>();
                
                operators = new ArrayList<Operator>();
                operators_selectivity = new ArrayList<Integer>();
                
                nbr_operator_call = 0;
                process = false;
                
                sel_sche_met = 0;
        }
        
        
        /**
         * Add operators
         */
        public void addOperator(Operator new_op)
        {
                new_op.setId( operators.size() );
                new_op.setParent(this);
                operators.add(new_op);
                operators_selectivity.add( new Integer(0) );
               
        }
        
        /**
         * Add a tuple
         * Caution with concurrent access !
         */
        public void addTuple( Tuple T )
        {
                T.setToDoBitmap(computeToDoBitmap(T));
                
                //Throwing the bugged tuple
                if(T.getToDoBitmap().cardinality() == 0)
                { System.out.println("Throwing unusable tuple : "+T.getTables()+" "+T.getColumns()); return; }
                
                input.add(T);
        }
        
        /**
         * Alternative addition to deals with selectivity
         * Should be used only by the operators
         * @param T Tuple to add
         * @param id Operator returning it
         */
        public void getBackTuple( Tuple T, int id )
        {               
                //Check if ended or not
                if(T.getDoneBitmap().cardinality() == operators.size())
                {
                        output.add(T);
                }               
                //If not, update bitmap and add it to the input queue
                else
                {
                        T.setToDoBitmap(computeToDoBitmap(T));
                        
                        //Throwing the bugged tuple
                        if(T.getToDoBitmap().cardinality() == 0)
                        {  System.out.println("Throwing unusable tuple : "+T.getTables()+" "+T.getColumns()); return; }
                        
                        input.add(T);                   
                }
                
                //Decreasing the operating selectivity since we got one back
                operators_selectivity.set(id, operators_selectivity.get(id) - 1);
        }
        
        /**
         * Set the bitmap to do, that is to say operator the tuple can go through.
         * It take a tuple and take it's owning table and compare it to the available.
         * The operators table should be included in the tuple tables
         * 
         */
        private BitSet computeToDoBitmap( Tuple T )
        {
                BitSet newToDoBM = new BitSet();
                BitSet doneBitmap = T.getDoneBitmap();

                boolean allFound = false; //All operator found, we do multiple 'and' with tabFound                
               
                //Go through all the operator
                for(int i=0; i<operators.size(); i++)
                {
                        //For each operator check if the Tables contains one of the table to deal with
                        ArrayList<String>  opTables = operators.get(i).getTables();
                        
                        allFound = false;
                                                
                        for(int j=0; j<opTables.size(); j++)
                        {
                                if( T.getTables().contains(opTables.get(j)) )
                                {
                                        allFound = true; //If one if false, one is missing we cannot do it
                                }
                        }
                        
                        //Add true if found and to already done
                        newToDoBM.set(i, (allFound && !doneBitmap.get(i)) );                    
                        
                }
                
                
                if(newToDoBM.cardinality() == 0)
                {
                        //Houston, we have a problem
                        System.out.print("No operator available !\n");  
                }
                
                return newToDoBM;
        }
        
        /**
         * Get the next operator to call.
         * It takes the next one in the array list.
         * @param to_do_bitmap Bitmap of the current analyzed tuple
         * @return Next operator to send the tuple
         */
        private int nextOperatorOrdered( BitSet to_do_bitmap )
        {
                return to_do_bitmap.nextSetBit(0);
        }
        
        /**
         * Get the next operator to call.
         * Take a random operator.
         * @param to_do_bitmap Bitmap of the current analyzed tuple
         * @return Next operator to send the tuple
         */
        private int nextOperatorRandom( BitSet to_do_bitmap )
        {
                int ret = -1;
                
                for(int i=0; i<Math.random()*100; i++)
                {
                        ret = to_do_bitmap.nextSetBit( i % to_do_bitmap.length() );
                }
                
                if(ret == -1)
                {
                        ret = to_do_bitmap.nextSetBit(0);
                }
                
                return ret;
        }
        
        /**
         * Get the next operator to call.
         * Use the naive algorithm based of performance.
         * @param to_do_bitmap Bitmap of the current analyzed tuple
         * @return Next operator to send the tuple
         */
        private int nextOperatorNaive( BitSet to_do_bitmap )
        {
                //If not enough operations done, call random one
                if(nbr_operator_call < min_nbr_op)
                { return nextOperatorRandom(to_do_bitmap); }
                
                int op_pos = 0;
                int ret = -1; //current select operator
                int cur_perf;
                
                op_pos = to_do_bitmap.nextSetBit(0);
                ret = op_pos;
                cur_perf = operators.get(ret).getPerfInd();
                
                while( op_pos != -1 )
                {
                        op_pos = to_do_bitmap.nextSetBit(op_pos+1);
                        
                        if( op_pos != -1 && operators.get(op_pos).getPerfInd() < cur_perf) //better cost
                        {
                                ret = op_pos;
                                cur_perf = operators.get(op_pos).getPerfInd();
                        }
                }
                
                return ret;
        }
        
        /**
         * Get the next operator to call.
         * Use the fast algorithm based on selectivity analysis
         * @param to_do_bitmap Bitmap of the current analyzed tuple
         * @return Next operator to send the tuple
         */
        private int nextOperatorFast( BitSet to_do_bitmap )
        {
                //If not enough operations done, call random one
                if(nbr_operator_call < min_nbr_op)
                { return nextOperatorRandom(to_do_bitmap); }
                
                ArrayList<Integer> op_max_val = new ArrayList<Integer>(); //For each operator, we set the max value, if the lottery have a lower value we take it
                int ticket_number = 0; //Total number of ticket in game (lottery)
                
                //Setting it...
                for(int i=0; i<operators.size(); i++)
                {
                		//If this is an available operator
                		if(to_do_bitmap.get(i))
                		{
	                		//Compute its importance (number of tickets) Selectivity less a ratio based on its cost
	                        int tickets = operators_selectivity.get(i) - operators.get(i).getPerfInd();
	                       
	                        if(tickets > 0)
	                        { ticket_number = ticket_number + tickets; }
	                        else //Minimum chance = 1
	                        { ticket_number++; }
	                        
	                        op_max_val.add(ticket_number);
                		}
                		else //Operator should not be taken
                		{
                			op_max_val.add(0); 
                		}
                }
                
                
                //Lottery !
                int lottery = (int) (Math.random() * ticket_number);
                
                //Go through, if the lottery is inferior, return
                for(int i=0; i<op_max_val.size(); i++)
                {
                        if(lottery < op_max_val.get(i))
                        { return i; }
                }
                
                return 0;
        }
        
        /**
         * Running and deal with the inputs one by one
         */
        @Override
        public void run() {
                                
                //Launching operator threads
                for(int i=0; i<operators.size(); i++)
                {
                        new Thread(operators.get(i)).start();
                }
                
                //Main loop
                process = true;
                
                while(process)
                {
                        //If there is an input, deal with it
                        if(input.size() > 0)
                        {       
                                Tuple tup_to_deal_with = input.poll();
                                
                                //Select the next operator method
                                int selected_operator = 0;
                                
                                if(sel_sche_met == 1)
                                { selected_operator = nextOperatorRandom( tup_to_deal_with.getToDoBitmap() );}
                                else if (sel_sche_met == 2)
                                { selected_operator = nextOperatorNaive( tup_to_deal_with.getToDoBitmap() );}
                                else if (sel_sche_met == 3)
                                { selected_operator = nextOperatorFast( tup_to_deal_with.getToDoBitmap() );}
                                else
                                { selected_operator = nextOperatorOrdered( tup_to_deal_with.getToDoBitmap() );}
                                
                                
                                System.out.println(tup_to_deal_with.getTables()+" "+selected_operator);
                                
                                //Sending to the operator
                                operators.get(selected_operator).addTuple(tup_to_deal_with);
                                operators_selectivity.set(selected_operator, operators_selectivity.get(selected_operator) + 1);
                                nbr_operator_call++;  
                               
                        }
                }               
                
                
        }
        
        
        /**
         * Get the output queue
         * @return the results
         */
        public LinkedList<Tuple> getOutput()
        {
                return output;          
        }
        
        
        /**
         * Select the scheduling methods
         */
        public void selectScheduleMethod(int sel)
        {
                if(0 < sel && sel < 4)
                { sel_sche_met = sel; }
        }
        
        
        /**
         * Get selectivity of the wanted operator.
         * @see operators_selectivity
         * @param op_pos Position of the operator in the array
         * @return The wanted value from operators_selectivity
         */
        public int getOperatorSelectivity(int op_pos)
        {
                return operators_selectivity.get(op_pos);
        }
        
        /**
         * Get operator performance
         * @param op_pos Position of the operator in the array
         * @return The operator performance index
         */
        public int getOperatorPerformance(int op_pos)
        {
                return operators.get(op_pos).getPerfInd();
        }
        
        /**
         * Get the number of operation called
         * @see nbr_operator_call
         * @return nbr_operator_call
         */
        public int getNumberOperatorCalled()
        {
                return nbr_operator_call;
        }
        
        /**
         * Get the wanted operator
         * @param op_pos Position of the operator in the array
         * @return the operator
         */
        public Operator getOperator(int op_pos)
        {
                return operators.get(op_pos);
        }
}

